package test.zipParser;
import java.io.IOException;
import java.util.List;
import java.sql.Time;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

import com.bj58.data.track.utility.format.TrackInputFormat;
import com.bj58.data.track.utility.info.TrackInfo;
import com.bj58.data.track.utility.tool.TrackObjectFormat;
import com.bj58.data.track.utility.tool.TrackSplitUtils;
//lib: com.bj58.data.utility-0.0.1-SNAPSHOT.jar

/**
 * ClassName: TrackinfoDemo <br/>
 * Function: 按时间段分城市统计二手车列表页PV，详情页PV，人均PV，信息量 <br/>
 * date: 2013-07-16 <br/>
 *
 * @author zhanghao06
 * @version 
 * @since JDK 1.6
 */


public class TrackinfoDemo
{
	public static class ZhMap extends Mapper<LongWritable, Text, Text, LongWritable>
	{
		private Text outkey =new Text();
		private LongWritable outvalue =new LongWritable();
		
		TrackSplitUtils trackSplit = TrackSplitUtils.getInstance();
		TrackInfo trackinfo;
		
		public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException 
		{
			List<List<String>> listLines = trackSplit.getSplitVisit(value);
			
			for(List<String> listline : listLines)
			{
				for(String line: listline)
				{
					//注意用continue，别用return
					trackinfo = TrackObjectFormat.getTrackinfoByString(line);
					
					String cate = new String(trackinfo.getTkCatePath());
					String area = trackinfo.getTkAreaPath();
					double dt = Double.parseDouble(trackinfo.getTkEpoch());
					
					long lt = (long)dt;
					Time time = new Time(lt);
					String curtime =new String(time.toString());
					
					outkey.set(cate+"_"+area + "_" +curtime);
					outvalue.set(1);
					context.write(outkey, outvalue);
				}
			}
		}	
	}
	
	
	public static class ZhReduce extends Reducer<Text, LongWritable, Text, LongWritable>
	{
		private LongWritable outval = new LongWritable();
		
		public void reduce(Text key, Iterable<LongWritable> values, Context context) throws IOException, InterruptedException 
		{
			long sum = 0;
			for(LongWritable val: values)
			{
				sum += val.get();
			}
			outval.set(sum);
			context.write(key, outval);
		}
	}
	
	public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException
	{
		Configuration conf = new Configuration();
		conf.set("mapred.job.queue.name", "regular");
		String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();

		if (otherArgs.length != 1) 
		{
			System.out.println("************************************************");
			System.out.println("************************************************");
			System.out.println("Usage: please input 1 params, for example: file.jar args[0]");
			System.out.println("args[0] is dateList: 20130101,20130106 or 20130106");
			System.out.println("************************************************");
			System.out.println("************************************************");
			System.exit(127);
		}

		String startDate = otherArgs[0].split(",")[0];
		String endDate = otherArgs[0].split(",").length == 2 ? otherArgs[0].split(",")[1] : startDate;
		String baseInPath = "/dsap/rawdata/track.58.com";
		String baseOutPath = "/dsap/middata/zhanghao06/TrackinfoDemo";
		int returnCode = 126;

		for (String runDate : MyDateUtil.getDateList(startDate, endDate, 0)) 
		{
			conf.set("DATE", runDate);
			Job job = new Job(conf, "TrackinfoDemo");
			job.setInputFormatClass(TrackInputFormat.class);
			job.setJarByClass(TrackinfoDemo.class);
			job.setMapperClass(ZhMap.class);
			job.setReducerClass(ZhReduce.class);
			job.setNumReduceTasks(1);
			job.setOutputKeyClass(Text.class);
			job.setOutputValueClass(LongWritable.class);
			job.setMapOutputKeyClass(Text.class);
			job.setMapOutputValueClass(LongWritable.class);

			if(FileSystem.get(conf).exists(new Path(baseInPath + "/" + runDate)))
			{
				FileInputFormat.addInputPath(job, new Path(baseInPath + "/" + runDate));
			}
			else
			{
				continue;
			}
			
			String outPath = baseOutPath + "/" + runDate;
			FileSystem.get(conf).delete(new Path(outPath), true);
			FileOutputFormat.setOutputPath(job, new Path(outPath));
			returnCode = job.waitForCompletion(true) ? 0 : 1;
		}

		System.exit(returnCode);
		
	}
	
}